package com.smsc.headend.task.engine.message;

import cn.hutool.core.date.SystemClock;
import cn.hutool.json.JSONUtil;
import com.google.common.base.Preconditions;
import com.google.protobuf.InvalidProtocolBufferException;
import com.smsc.headend.common.utils.ProtoBufferUtil;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.proto.TaskProto;
import com.smsc.headend.task.engine.config.KafkaConfig;
import com.smsc.headend.task.engine.service.task.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.commons.lang.time.DateUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.*;

@Component
@Slf4j
public class DelayQueueMessage {

    public static final String DELAY_MIN_15 = "delayMin15";
    public static final String DELAY_MIN_60 = "delayMin60";
    public static final String DELAY_HOUR_4 = "delayMin240";

    @Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;
    @Autowired
    TaskService taskService;

    @KafkaListener(id = DELAY_MIN_15, idIsGroup = false, topics = TaskKafkaTopic.TASK_DELAY15_TO_UTE, containerFactory = KafkaConfig.ONE_CONSUME_FACTORY, autoStartup = TaskMessageService.FALSE)
    public void receiveFifteenMinuteDelayMessage(List<ConsumerRecord<String, byte[]>> dataList, Acknowledgment acknowledgment) {
        ConsumerRecord<String, byte[]> consumerRecord = dataList.get(0);
        Task task = acquireTask(consumerRecord);
        Preconditions.checkNotNull(task);
        Long sendTimestamp = consumerRecord.timestamp();
        log.info("taskNo={}, taskType={}, sendTime ={}", task.getTaskNo(), task.getTaskType(), DateFormatUtils.format(new Date(sendTimestamp), "yyyy-MM-dd HH:mm:ss"));
        Long scheduleTimestamp = sendTimestamp + 15 * DateUtils.MILLIS_PER_MINUTE;

        if (scheduleTimestamp < SystemClock.now()) {
            DelayQueueMessage.this.executeTask(task, acknowledgment, DELAY_MIN_15, false);
        } else {
            MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer(DELAY_MIN_15);
            listenerContainer.pause();
            setTimerForExecuteTask(task, scheduleTimestamp, acknowledgment, DELAY_MIN_15);
        }
    }


    @KafkaListener(id = DELAY_MIN_60, idIsGroup = false, topics = TaskKafkaTopic.TASK_DELAY60_TO_UTE, containerFactory = KafkaConfig.ONE_CONSUME_FACTORY, autoStartup = TaskMessageService.FALSE)
    public void receiveOneHourDelayMessage(List<ConsumerRecord<String, byte[]>> dataList, Acknowledgment acknowledgment) {
        ConsumerRecord<String, byte[]> consumerRecord = dataList.get(0);
        Task task = acquireTask(consumerRecord);
        Preconditions.checkNotNull(task);
        Long sendTimestamp = consumerRecord.timestamp();
        log.info("taskNo={}, sendTime ={}", task.getTaskNo(), DateFormatUtils.format(new Date(sendTimestamp), "yyyy-MM-dd HH:mm:ss"));
        Long scheduleTimestamp = sendTimestamp + DateUtils.MILLIS_PER_HOUR;

        if (scheduleTimestamp < SystemClock.now()) {
            DelayQueueMessage.this.executeTask(task, acknowledgment, DELAY_MIN_60, false);
        } else {
            MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer(DELAY_MIN_60);
            listenerContainer.pause();
            setTimerForExecuteTask(task, scheduleTimestamp, acknowledgment, DELAY_MIN_60);
        }
    }

    @KafkaListener(id = DELAY_HOUR_4, idIsGroup = false, topics = TaskKafkaTopic.TASK_DELAY240_TO_UTE, containerFactory = KafkaConfig.ONE_CONSUME_FACTORY, autoStartup = TaskMessageService.FALSE)
    public void receiveFourHourDelayMessage(List<ConsumerRecord<String, byte[]>> dataList, Acknowledgment acknowledgment) {
        ConsumerRecord<String, byte[]> consumerRecord = dataList.get(0);
        Task task = acquireTask(consumerRecord);
        Preconditions.checkNotNull(task);
        Long sendTimestamp = consumerRecord.timestamp();
        log.info("taskNo={}, sendTime ={}", task.getTaskNo(), DateFormatUtils.format(new Date(sendTimestamp), "yyyy-MM-dd HH:mm:ss"));
        Long scheduleTimestamp = sendTimestamp + 4 * DateUtils.MILLIS_PER_HOUR;
        if (scheduleTimestamp < SystemClock.now()) {
            DelayQueueMessage.this.executeTask(task, acknowledgment, DELAY_HOUR_4, false);
        } else {
            MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer(DELAY_HOUR_4);
            listenerContainer.pause();
            setTimerForExecuteTask(task, scheduleTimestamp, acknowledgment, DELAY_HOUR_4);
        }
    }

    private void setTimerForExecuteTask(Task task, Long scheduleTimestamp, Acknowledgment acknowledgment, String kafkaContainerId) {
        Timer timer = new Timer();
        log.info("task {} will be execute at {}", task.getTaskNo(), DateFormatUtils.format(new Date(scheduleTimestamp), "yyyy-MM-dd HH:mm:ss"));
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                DelayQueueMessage.this.executeTask(task, acknowledgment, kafkaContainerId, true);
            }
        }, new Date(scheduleTimestamp));
    }

    private Task acquireTask(ConsumerRecord<String, byte[]> consumerRecord) {
        TaskProto.Task task = null;
        try {
            task = TaskProto.Task.parseFrom((byte[]) consumerRecord.value());
        } catch (InvalidProtocolBufferException e) {
            log.error("task data error");
        }
        Task incomeTask = ProtoBufferUtil.fromProto(task, Task.class);
        return incomeTask;
    }

    private void executeTask(Task task, Acknowledgment acknowledgment, String kafkaContainerId, boolean resumeConsume) {
        log.info("try start task here {}", task);
        try {
            taskService.sendTaskToGenerator(task);
        } catch (Exception e) {
            log.error("execute failed, {}", JSONUtil.toJsonStr(task), e);
        }
        MessageListenerContainer listenerContainer = kafkaListenerEndpointRegistry.getListenerContainer(kafkaContainerId);
        if (resumeConsume) {
            listenerContainer.resume();
        }
        acknowledgment.acknowledge();
    }

}
